package cool.taomu.toolkit.service.client.kademlia.service

import com.google.common.base.Optional
import com.google.common.util.concurrent.FutureCallback
import com.google.gson.Gson
import com.google.gson.reflect.TypeToken
import com.google.inject.Inject
import com.google.inject.name.Named
import cool.taomu.toolkit.Context
import cool.taomu.toolkit.executor.TaomuExecutors
import cool.taomu.toolkit.service.kademlia.entity.KPeerNode
import cool.taomu.toolkit.service.kademlia.service.ICheckPeer
import cool.taomu.toolkit.service.kademlia.service.KRoutingTable
import cool.taomu.toolkit.service.kademlia.thrift.Node
import cool.taomu.toolkit.service.kademlia.thrift.Response
import cool.taomu.toolkit.service.kademlia.thrift.ServiceEntity
import cool.taomu.toolkit.utils.ByteBufferUtils
import cool.taomu.toolkit.utils.StringUtils
import java.util.ArrayList
import java.util.List
import java.util.concurrent.Callable
import java.util.concurrent.SynchronousQueue
import java.util.concurrent.TimeUnit
import org.eclipse.xtend.lib.annotations.Accessors
import org.slf4j.LoggerFactory

class ServiceCentreService {

    val static LOG = LoggerFactory.getLogger(ServiceCentreService)

    @Inject
    FindServiceImpl findService;

    @Inject
    @Named("taomu.kademlia.peer.host")
    String host

    @Inject
    @Named("taomu.kademlia.peer.port")
    int port;

    @Inject
    KRoutingTable router;

    @Inject
    ICheckPeer check

    @Inject
    Context context

    @Accessors
    static class ServiceNode {
        Node node;
        ServiceEntity entity;

        new(Node node, ServiceEntity entity) {
            this.node = node
            this.entity = entity
        }
    }

    def asyncCallService(List<Optional<ServiceNode>> result) {
        var List<Callable<Optional<Response>>> calls = result.map [
            [
                if (it.present) {
                    var snode = it.get() as ServiceNode
                    var res = findService.callService(new KPeerNode(snode.node.host, snode.node.port), snode.entity)
                    if (res.present) {
                        LOG.info("callService 线程 执行 返回的结果 :{}", res.get())
                        return res
                    }
                }
                return Optional.absent
            ]
        ]
        LOG.debug("Callable数量:{}", calls.size)
        val responses = new SynchronousQueue<ArrayList<Optional<Response>>>()
        var executors = TaomuExecutors.getInstance(this.context.executor)
        executors.tasks(calls).allAsList.callback(new FutureCallback<List<Optional<Response>>>() {
            override onFailure(Throwable t) {
                LOG.debug("返回失败了:", t)
            }

            override onSuccess(List<Optional<Response>> result) {
                if (result.size > 0) {
                    var list = new ArrayList<Optional<Response>>()
                    list.addAll(result)
                    responses.add(list)
                }
            }
        })
        responses.take
    }

    def finedServices(List<ServiceEntity> serviceEntityList) {
        LOG.info("****finedServices****")
        LOG.info("****K桶中节点数量:{}****", this.router.ksize)
        val responses = new SynchronousQueue<ArrayList<Optional<Response>>>()
        var executors = TaomuExecutors.getInstance(this.context.executor)
        executors.tasks(this.createCallable(serviceEntityList)).allAsList.callback(
            new FutureCallback<List<Optional<ServiceNode>>>() {
                override onFailure(Throwable t) {
                    LOG.debug("返回失败了:", t)
                }

                override onSuccess(List<Optional<ServiceNode>> result) {
                    if (result.size > 0) {
                        LOG.info("获取的节点的集合数:{}", result.size)
                        var list = new ArrayList<Optional<Response>>()
                        list.addAll(asyncCallService(result))
                        responses.add(list)
                    }
                }

            }
        )
        responses.take
    }

    def List<Callable<Optional<ServiceNode>>> createCallable(ServiceEntity ... serviceEntitys) {
        return serviceEntitys.map [
            val key = #[it.method, it.serviceName].join(":")
            return [
                val key160 = StringUtils.stringTo160Bit(key)
                var klist = router.findClosestNodes(key160, 8)
                LOG.info("根据key160 {} 找到的节点数:{}", key160, klist.size)
                var peer = new KPeerNode(host, port);
                if (klist !== null && klist.size > 0) {
                    var host = klist.get(0).host
                    var port = klist.get(0).port
                    peer = new KPeerNode(host, port);
                }
                // TODO 需要优化
                var res = findService.getServiceNodes(peer, it, new ArrayList());
                if (res.present) {
                    var result = res.get();
                    if (result.status == 200) {
                        var dataStr = ByteBufferUtils.toString(result.data, "UTF-8")
                        var nodes = new Gson().fromJson(dataStr, new TypeToken<List<KPeerNode>>() {
                        })
                        LOG.info("找到服务节点数量:{}\n==>{}", nodes.size, nodes)
                        nodes.forEach [
                            // if(check.apply(new KPeerNode(it.host, it.port))){
                            this.context.executor.submit([
                                router.addPeerNode(it, check)
                            ])
                        // }
                        ]
                        var node = new Node(nodes.get(0).host, nodes.get(0).port)
                        return Optional.fromNullable(new ServiceNode(node, it))
                    } else {
                        var node = new Node(peer.host, peer.port)
                        return Optional.fromNullable(new ServiceNode(node, it))
                    }
                } else {
                    LOG.info("没有找到节点")
                    return Optional.absent;
                }
            ]
        ]
    }

    def Response finedService(ServiceEntity serviceName) {
        LOG.info("****finedService****: \n{}", serviceName)
        LOG.info("****K桶中节点数量:{}****", this.router.ksize)
        var key = #[serviceName.method, serviceName.serviceName].join(":")
        LOG.info("****finedService key ****: {}", key)
        val sq = new SynchronousQueue<Optional<Response>>()
        var executors = TaomuExecutors.getInstance(this.context.executor)
        executors.tasks(this.createCallable(serviceName)).allAsList.callback(
            new FutureCallback<List<Optional<ServiceNode>>>() {
                override onFailure(Throwable t) {
                    LOG.debug("返回失败了:", t)
                }

                override onSuccess(List<Optional<ServiceNode>> result) {
                    if (result !== null && result.size > 0) {
                        var snode = result.get(0)
                        if (snode.present) {
                            var node = new KPeerNode(snode.get.node.host, snode.get.node.port)
                            var r = findService.callService(node, snode.get().entity)
                            LOG.info("r.present : {} ", r.present)
                            sq.add(r)
                        }
                    }
                }
            })
        var res = sq.poll(1, TimeUnit.MINUTES)
        LOG.info("返回结果了！！！！！！！")
        if (res.present) {
            return res.get()
        }
        return new Response(ByteBufferUtils.toByteBuffer("404"), 404, "未找到方法");
    }

}
